package com.jiazhong.mingxing.boot.boot11.顺序消息;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.List;

@Slf4j
public class 顺序消息消费 {

    @SneakyThrows
    public static void main(String[] args) {
        final int[] i = {1};
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("jiazhong-boot-a");
        consumer.setNamesrvAddr("124.220.170.254:9876");
        consumer.subscribe("TopicC", "*");
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                i[0]++;
                try {
                    if (i[0] == 14) {
                        int i = 8 / 0;
                    }
                    log.info("message:{}", new String(list.get(0).getBody(), StandardCharsets.UTF_8));
                } catch (Exception e) {
                    log.error("出现了异常");
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
    }
}
